package spring.hadoopdemo;

import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import sun.nio.ch.IOUtil;

import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;

/**
 * @author legolas
 * @date 2020/3/22 12:03
 * 使用Spring Hadoop 访问HDFS文件系统
 */
public class SpringHadoopHDFSApp {

    private ApplicationContext ctx;
    private FileSystem fileSystem;

    //创建hdfs文件夹
    @Test
    public void mkdir() throws Exception {
        fileSystem.mkdirs(new Path("/springhdfs"));
    }

    //读取hdfs文件内容
    @Test
    public void cat() throws Exception {
        FSDataInputStream in = fileSystem.open(new Path("/springhdfs/hello.txt"));
        IOUtils.copyBytes(in, System.out, 1024);
        in.close();
    }

    @Test
    public void rename() throws Exception {
        Path oldPath = new Path("/springhdfs/hello.txt");
        Path newPath = new Path("/springhdfs/hello1.txt");
        fileSystem.rename(oldPath, newPath);
    }


    //上传文件到HDFS
    @Test
    public void copyFromLocalFile() throws Exception {
        Path localPath = new Path("/springhdfs/hello.txt");
        Path hdfsPath = new Path("/springhdfs/hello.txt");
        fileSystem.copyFromLocalFile(localPath, hdfsPath);
    }

    //上传文件到HDFS(带进度提醒)
    @Test
    public void copyFromLocalFileWithProgress() throws Exception {
        InputStream in = new BufferedInputStream(new FileInputStream(new File("/springhdfs/hello.txt")));
        FSDataOutputStream outputStream = fileSystem.create(new Path("/springhdfs"),
                () -> {
                    System.out.print(".");//带进度提醒信息
                });
        IOUtils.copyBytes(in, outputStream, 4096);
    }

    //下载HDFS文件
    @Test
    public void copyToLocalFile() throws Exception {
        Path localPath = new Path("/springhdfs/hello.txt");
        Path hdfsPath = new Path("/springhdfs/hello.txt");
        fileSystem.copyToLocalFile(hdfsPath, localPath);
    }

    //遍历目录下所有文件
    @Test
    public void listFiles() throws Exception {
        FileStatus[] fileStatuses = fileSystem.listStatus(new Path("/"));
        for (FileStatus fileStatus : fileStatuses) {
            String isDir = fileStatus.isDirectory() ? "文件夹" : "文件";
            short replication = fileStatus.getReplication();
            long len = fileStatus.getLen();
            String path = fileStatus.getPath().toString();
            System.out.println(isDir + "\t" + replication + "\t" + len + "\t" + path);
        }
    }

    @Before
    public void setUp() {
        ctx = new ClassPathXmlApplicationContext("beans.xml");
        fileSystem = (FileSystem) ctx.getBean("fileSystem");
    }

    @After
    public void tearDown() throws Exception {
        ctx = null;
        fileSystem.close();
    }

}
